package com.yujiahao.bigdata.acc

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark01_Acc_WordCount {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接
    // 参数解读：setAppName("当前执行程序的名字").setMaster("当前的环境")
    // 分布式环境的基本架构为 主 - 从
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)
    //TODO 3、业务逻辑 --自定义累加器
    val rdd: RDD[(String)] = sc.makeRDD(List("hello", "word", "spark", "hello", "Scala"), 2)

    //累加器没有shuffle
    //使用累加器步骤
    // TODO 3.1）声明累加器
    val wordCount: WordcountAccumulator = new WordcountAccumulator()
    // TODO 3.2）向Spark注册累加器，要和Spark取得联系
    sc.register(wordCount)

    //TODO 3.3）下面将数据进行遍历累加
    rdd.foreach(
      word => {
        //TODO 3.4）这里真正的将遍历出来的单词放在累加器当中
        wordCount.add(word)
      }
    )
    //4、获取累加器的结果
    val result: mutable.Map[String, Int] = wordCount.value
    println(result)

  }


  //TODO WordCount累加器--累加器没有shuffle的概念
  //  1、继承AccumulatorV2
  //  2、定义泛型：
  //  IN:输入的数据类型--这里的话输入的是String类型
  //  OUT:输出的数据类型--输出的是勒可变集合的Map键值对
  //  3、重写方法
  class WordcountAccumulator extends AccumulatorV2[String, mutable.Map[String, Int]] {
    private val wcMap = mutable.Map[String, Int]()

    //累加器是否为初始化状态
    override def isZero: Boolean = {
      wcMap.isEmpty
    }

    //复制累加器
    override def copy(): AccumulatorV2[String, mutable.Map[String, Int]] = {
      new WordcountAccumulator
    }

    //重置累加器
    override def reset(): Unit = {
      wcMap.clear()
    }

    //累加数据
    override def add(word: String): Unit = {
      val oldCnt: Int = wcMap.getOrElse(word, 0)
      wcMap.update(word, oldCnt + 1)
    }

    //合并累加器
    override def merge(other: AccumulatorV2[String, mutable.Map[String, Int]]): Unit = {
      val map2 = other.value
      map2.foreach {
        case (word, count) => {
          val getCnt: Int = wcMap.getOrElse(word, 0)
          wcMap.update(word, count + getCnt)
        }
      }

    }

    //返回累加结果
    override def value: mutable.Map[String, Int] = wcMap
  }

}
